# -*- coding: UTF-8 -*-
"""
@File    ：3.数据计算.py
@Author  ：techPang
@Date    ：2025/2/8 14:30 
@Version ：1.0
"""
from pyspark import SparkConf, SparkContext

# 创建sparkConf类对象
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")

# 基于配置类创建SparkContext对象
sc = SparkContext(conf=conf)
# -------------------------------------------map-------------------------------------------------------
# 1.map方法，把rdd中的每一条数据拿出来进行处理
# 准备一个数据
# rdd = sc.parallelize([1, 2, 3, 4, 5])

# 对rdd中的数据每条都乘以10
# def func_map(data):
#     return data * 10

# 使用步骤:在map中传入一个方法，必须有一个入参和一个出参，支持链式调用    (T) -> U  传入一个值，返回一个值，类型没有要求
# rdd_map = rdd.map(lambda x: x * 10).map(lambda x: x + 10)
# print(rdd_map.collect())

# -----------------------------------------flatMap-----------------------------------------------------
# 2.flatMap，对rdd进行map操作，然后进行解除嵌套操作
# lst = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]  解除嵌套后:list = [1, 2, 3, 4, 5, 6, 7, 8, 9]

# 准备一个数据
# rdd = sc.parallelize(["czt 23 man", "zh 22 woman", "ljx, 23, man"])

# rdd_flatMap = rdd.flatMap(lambda x: x.split(" "))
# rdd_map = rdd.map(lambda x: x.split(" "))

# 输出对比
# print(rdd_map.collect())       [['czt', '23', 'man'], ['zh', '22', 'woman'], ['ljx,', '23,', 'man']]
# print(rdd_flatMap.collect())   ['czt', '23', 'man', 'zh', '22', 'woman', 'ljx,', '23,', 'man']

# ---------------------------------------reduceByKey---------------------------------------------------
# 3.reduceByKey，针对KV型RDD，自动按照key分组，然后根据你提供的聚合逻辑，完成组内数据(value)的聚合操作
# (V,V) -> V 传入的两个值和输出的值类型统一
# 传入二元元组(元组中只有两个值，前面的值为K，后面的值为V)

# 准备一个数据
# rdd = sc.parallelize([('a', 1), ('a', 2), ('a', 2), ('b', 1), ('b', 3)])

# rdd_reduced = rdd.reduceByKey(lambda x, y: x + y)
# print(rdd_reduced.collect())

# -------------------------------------------filter-----------------------------------------------------
# 过滤想要保留的数据  (T) -> bool，为True的保留，为False的去掉

# 准备一个数据
# rdd = sc.parallelize([1, 2, 3, 4, 5])

# 只要偶数
# rdd_f = rdd.filter(lambda num: num % 2 == 0)
# print(rdd_f.collect())

# -------------------------------------------distinct-----------------------------------------------------
# 对数据进行去重操作
# 准备一个数据
# rdd = sc.parallelize([1, 2, 2, 3, 4, 4, 5])
#
# 进行去重操作
# rdd_d = rdd.distinct()
# print(rdd_d.collect())

# -------------------------------------------sortBy-----------------------------------------------------
# 对数据进行排序，需要指定规则 (T) -> U
# 接收三个参数 rdd.sortBy(func, ascending=False(True为升序，False为降序), numPartitions=1(分区))
# 实例见案例1最后的排序操作

# 关闭
sc.stop()
